package producer

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
	"go_knowledge/mq/kafka/event"
	"go_knowledge/tool/utils"
	"time"
)

type defaultSyncClient struct {
	addrs  []string
	topic  string
	logger *logrus.Logger

	failedHandler event.PushFailedHandler
	client        *sarama.SyncProducer
	config        *sarama.Config
}

func NewDefaultSyncProducerClient(addrs []string, topic string) *defaultSyncClient {

	config := sarama.NewConfig()
	config.Version = sarama.V0_10_2_0
	//config.Version = sarama.V2_1_0_0
	config.Producer.Return.Successes = true
	//config.ClientID = topic

	client := &defaultSyncClient{
		addrs:  addrs,
		topic:  topic,
		config: config,
	}
	//client.start()

	return client
}

func (dpc *defaultSyncClient) SetLogger(logger *logrus.Logger) {
	dpc.logger = logger
}

func (dpc *defaultSyncClient) SetTopic(topic string) {
	if topic == "" {
		dpc.topic = topic
	}
}

func (dpc *defaultSyncClient) SetFailedHandler(h event.PushFailedHandler) {
	dpc.failedHandler = h
}

func (dpc *defaultSyncClient) start() {
	// 如果未设置 logger，默认初始化一个
	if nil == dpc.logger {
		l := logrus.New()
		l.SetLevel(logrus.DebugLevel)
		dpc.logger = l
		fmt.Println("not set logger, use default &clog.Logger{}")
	}

	// 启动 producer
	producer, err := sarama.NewSyncProducer(dpc.addrs, dpc.config)
	if err != nil {
		dpc.logger.Errorf("creating sync producer client failed: %v", err)
		return
	}

	dpc.client = &producer
}

func (dpc *defaultSyncClient) PushPbEvent(ev event.PbEvent, blockKey string) error {
	if nil == ev {
		return errors.New("msg is nil")
	}

	data, err := ev.XXX_Marshal(nil, false)
	if nil != err {
		return errors.Wrap(err, "event marshal failed")
	}

	// json 序列化
	msg := &event.PbEventMessage{
		Type: utils.FullTypeNameOf(ev),
		Data: data,
	}
	bt, err := json.Marshal(msg)
	if nil != err {
		return errors.Wrap(err, "PbEventMessage to bytes failed")
	}

	return dpc.sendMessage(bt, blockKey)
}

func (dpc *defaultSyncClient) sendMessage(msg []byte, blockKey string) error {
	// 延迟启动
	dpc.start()

	if len(msg) == 0 {
		return errors.New("msg is nil")
	}

	if nil == dpc.client {
		return errors.New("no producer client")
	}

	data := &sarama.ProducerMessage{
		Topic: dpc.topic,
		Key:   sarama.StringEncoder(blockKey),
		Value: sarama.ByteEncoder(msg),
	}

	interval := 200 * time.Millisecond
	tick := time.NewTicker(interval)
	defer tick.Stop()

	toCtx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()

	var err error
ForLoop:
	for {
		select {
		case <-tick.C:
			err = (*dpc.client).SendMessages([]*sarama.ProducerMessage{data})
			if err == nil {
				dpc.logger.Debug("send message success")
				break ForLoop
			}
		case <-toCtx.Done():
			err = errors.New("send message timeout")
			break ForLoop
		}
	}

	return err
}

func (dpc *defaultSyncClient) Close() error {
	if nil != dpc.client {
		return (*dpc.client).Close()
	}

	return nil
}
